package com.heima.article.stream;
import com.alibaba.fastjson.JSON;
import com.heima.common.constant.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.time.Duration;
/**
 * 实时更新热点文章
 */
@Configuration
@Slf4j
public class HotArticleStreamHandler {
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        log.info("【实时更新热点文章】--- begin");
        //设置监听主题
        KStream<String, String> kStream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        kStream.map((key, value) -> {
            //value -> mess
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            //重置消息 key:articleId 和 value:like:1
            return new KeyValue<>(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
        })
            //按照key聚合
            .groupBy((key,value)->key)
            //时间窗口
            .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
            //自定义聚合计算
            .aggregate(new Initializer<String>() {
               /**
                * 初始方法，返回值是消息的value
                *
                * @return
                */
               @Override
               public String apply() {
                   return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
               }
           }, new Aggregator<String, String, String>() {
                    /**
                     * 真正的聚合操作，返回值是消息的value
                     */
                   @Override
                   public String apply(String key, String value, String aggValue) {
                       if(StringUtils.isBlank(value)){
                           return aggValue;
                       }
                       //获取文章的 col com lik vie
                       int col=0,com=0,lik=0,vie=0;
                       String[] aggStr = aggValue.split(",");
                       for(String s:aggStr){
                           switch(s.split(":")[0]){
                               case "COLLECTION":col = Integer.parseInt(s.split(":")[1]);break;
                               case "COMMENT":com = Integer.parseInt(s.split(":")[1]);break;
                               case "LIKES":lik = Integer.parseInt(s.split(":")[1]);break;
                               case "VIEWS":vie = Integer.parseInt(s.split(":")[1]);break;
                           }
                       }
                       //更新文章评分
                       String[] values = value.split(":");
                       switch(values[0]){
                           case "COLLECTION":col += Integer.parseInt(values[1]);break;
                           case "COMMENT":com += Integer.parseInt(values[1]);break;
                           case "LIKES":lik += Integer.parseInt(values[1]);break;
                           case "VIEWS":vie += Integer.parseInt(values[1]);break;
                       }
                       System.out.println("文章id:"+key);
                       String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                       System.out.println("formatStr = " + formatStr);
                       return formatStr;
                   }
               }, Materialized.as("hot-article-stream-count-001")) //设置流式处理当前状态，任意字符串但不能重复
                    .toStream()

                    //发送消息
                    .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC, Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class)));
        log.info("【实时更新热点文章】--- end");
        return kStream;
    }

    /**
     * 格式化消息的value数据
     * @param articleId
     * @param value
     * @return
     */
    public String formatObj(String articleId,String value){
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(articleId));
        //COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
        String[] valAry = value.split(",");
        for (String val : valAry) {
            String[] split = val.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("聚合消息处理之后的结果为:{}",JSON.toJSONString(mess));
        return JSON.toJSONString(mess);

    }
}
